package com.nervenets.web.rabbitmq;

import com.nervenets.Global;
import com.nervenets.utils.JodaUtils;
import com.nervenets.utils.NumberUtil;
import com.nervenets.utils.StringUtils;
import com.nervenets.web.enumeration.*;
import com.nervenets.web.hibernate.dao.*;
import com.nervenets.web.hibernate.domain.Order;
import com.nervenets.web.hibernate.domain.WealthFlow;
import com.nervenets.web.hibernate.domain.knowledge.Tags;
import com.nervenets.web.redis.GlobalService;
import com.nervenets.web.redis.GlobalVariablesService;
import com.nervenets.web.redis.domain.user.UserStored;
import com.nervenets.web.redis.utils.ISyncExecutor;
import com.nervenets.web.redis.utils.RedisSyncLocker;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * Created by Joe on 2018/1/4.
 */
@Log4j
@Component
@RabbitListener(queues = "knowledge", containerFactory = "rabbitListenerContainerFactory")
public class FlowsQueueReceiver {
    private static Logger logger = LoggerFactory.getLogger(FlowsQueueReceiver.class);
    @Resource
    private RedisSyncLocker redisSyncLocker;
    @Resource
    private UserDao userDao;
    @Resource
    private OrderDao orderDao;
    @Resource
    private WealthFlowDao wealthFlowDao;
    @Resource
    private GlobalService globalService;
    @Resource
    private GlobalVariablesService globalVariablesService;
    @Resource
    private StatisticsOfUserDao statisticsOfUserDao;
    @Resource
    private StatisticsOfOrderDao statisticsOfOrderDao;
    @Resource
    private TagsDao tagsDao;
    @Resource
    private KnowledgeDao knowledgeDao;

    @RabbitHandler
    public void process(@Payload FlowMessage flowMessage, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws Exception {
        long userId = flowMessage.getUserId();
        long ip = flowMessage.getLongValue("ip");

        FlowType type = flowMessage.getFlowType();
        switch (type) {
            case userActive:
                activeUser(userId);
                channel.basicAck(deliveryTag, false);
                break;
            case wealthChange:
                long offset = flowMessage.getIntValue("offset");
                long unionId = flowMessage.getUnionId();
                WealthType wealthType = WealthType.money;
                if (!StringUtils.isEmpty(flowMessage.getStringValue("wealthType"))) {
                    wealthType = WealthType.valueOf(flowMessage.getStringValue("wealthType"));
                }
                OperateType operateType = OperateType.valueOf(flowMessage.getStringValue("operateType"));
                PayType payType = PayType.valueOf(flowMessage.getStringValue("payType"));
                UnionType unionType = UnionType.valueOf(flowMessage.getStringValue("unionType"));
                FlowStatus flowStatus = FlowStatus.valueOf(flowMessage.getStringValue("flowStatus"));
                String detail = flowMessage.getStringValue("detail");
                wealthChange(offset, userId, wealthType, operateType, detail, payType, ip, unionId, unionType, flowStatus);
                channel.basicAck(deliveryTag, false);
                break;
            case orderPaid:
                if (orderPaidHandler(flowMessage)) {
                    channel.basicAck(deliveryTag, false);
                }
                break;
            case tagsCount:
                tagsCountHandler(flowMessage);
                channel.basicAck(deliveryTag, false);
                break;
        }
    }

    private void tagsCountHandler(FlowMessage flowMessage) throws Exception {
        redisSyncLocker.idLock(flowMessage.getUnionId(), o -> {
            final Tags tags = tagsDao.findById(flowMessage.getUnionId()).orElse(null);
            if (null != tags) {
                tags.setContentCount(knowledgeDao.countByTagsId(tags.getId()));
                tagsDao.save(tags);
            }
            return null;
        });
    }

    private boolean orderPaidHandler(FlowMessage flowMessage) throws Exception {
        return (boolean) redisSyncLocker.idLock(flowMessage.getUnionId(), (ISyncExecutor<Boolean, Object>) o -> {
            try {
                final Order order = orderDao.findById(flowMessage.getUnionId()).orElse(null);
                if (null != order) {
                    if (order.getPayMoney() > 0) {
                        wealthChange(-order.getPayMoney(), order.getUserId(), WealthType.money, OperateType.service, "洗护服务", PayType.surplus, flowMessage.getLongValue("ip"), order.getId(), UnionType.order, FlowStatus.success);
                    }

                    order.addFlow(order.getStatus(), OrderStatus.paid, "下单成功，等待配送人上门取件", false);
                    order.setStatus(OrderStatus.paid);
                    order.setPayTime(JodaUtils.getTimestamp());
                    orderDao.save(order);
                    //统计
                    final long totalUser = orderDao.countByUserId(order.getUserId());
                    statisticsOfOrderDao.updateCount(1, order.getPayMoney(), 1 == totalUser ? 1 : 0, 0, 0, JodaUtils.getStartOfToday());
                    statisticsOfUserDao.updateCount(0, 0, 1, 0, JodaUtils.getStartOfToday());

                    //时间卡标记使用
                    final String cartLockedKey = String.format("CART%s", order.getUserId());
                    globalService.setGlobalPair(cartLockedKey, "");//清除购物车
                }
                return true;
            } catch (Exception e) {
                return false;
            }
        });
    }

    private void activeUser(long userId) {
        final UserStored user = globalService.getUser(userId);
        if (null != user) {
            final int startOfToday = JodaUtils.getStartOfToday();
            statisticsOfUserDao.updateCount(user.getCreateTime() > startOfToday ? 1 : 0, 1, 0, 0, startOfToday);
        }
    }

    private void wealthChange(long offset, long userId, WealthType wealthType, OperateType operateType, String detail, PayType payType, long ip, long unionId, UnionType unionType, FlowStatus flowStatus) throws Exception {
        if (0 == offset) return;
        redisSyncLocker.keyLock(String.format(Global.Constants.REDIS_USER_WEALTH_LOCK, userId), o -> {
            final UserStored user = globalService.getUser(userId);
            if (null == user) {
                logger.error(String.format("正在给未知用户【%s】进行账户变动", userId));
                return null;
            }

            WealthFlow lastFlow = wealthFlowDao.findTopByUserIdAndWealthTypeOrderByIdDesc(userId, wealthType);
            long surplus = null == lastFlow ? 0 : lastFlow.getSurplus();
            long total = (null == lastFlow ? 0 : lastFlow.getTotal());
            if (offset > 0 && OperateType.isTrueIncome(operateType)) {
                total += offset;
            }

            final long redisSurplus = globalService.getUserWealthOfMoney(userId, wealthType);
            if (surplus != redisSurplus) globalService.removeRedisWealthOfMoney(userId, wealthType);

            globalService.addUserWealthOfMoney(userId, wealthType, offset);
            globalService.addUserWealthOfMoneyByOperateType(userId, wealthType, operateType, offset);
            if (OperateType.isTrueIncome(operateType)) {
                globalService.addUserWealthOfTotalMoney(userId, wealthType, offset);
            }

            final long cashSurplus = surplus + offset;
            logger.info(String.format("用户%s上次结余%s元，%s变动%s后剩余%s元", userId, NumberUtil.fen2Yuan(surplus), operateType.getText(), NumberUtil.fen2Yuan(offset), NumberUtil.fen2Yuan(cashSurplus)));
            WealthFlow flow = new WealthFlow(userId, offset, cashSurplus, operateType, ip, unionId, unionType, detail);
            flow.setWealthType(wealthType);
            flow.setFlowStatus(flowStatus);
            flow.setPayType(payType);
            flow.setTotal(total);
            wealthFlowDao.save(flow);


            return null;
        }, null);
    }
}
